%%%-------------------------------------------------------------------
%%% File    : valg2pgt.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 13 May 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(valg2pgt).

-include("../include/valg.hrl").
-include("../include/vect.hrl").
-include("../include/rrs.hrl").
-include("../include/pipe.hrl").
-include("../include/iter.hrl").
-include("../include/pgt.hrl").
-include("../include/util.hrl").

-define(IS_RECFUN(Fun),is_function(Fun,1)).
-define(IS_VECT(Arg),is_record(Arg,vect)).

%% API
-export([exp_fold/2,fold/2,format/1,map_iter2pipe/3,as_is/1]).
-export([test/1]).

-record(acc,{
			 pg,
			 last_pipe,
			 pgt_dict,
			 route_snap=svc_rrs:make_route_snapshot()
			}).
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------

fold(Expr,OutPipe) ->
	MainPg={pg,make_ref(),"Evaluate expression"},
	Acc=exp_fold(Expr, 
				 #acc{
					  pg=MainPg,
					  last_pipe=OutPipe,
					  pgt_dict=dict:new()
					 }),
	#pg{
		id=erlang:make_ref(),
		name="Controller",
		pgs=[make_pgt(Acc#acc.pgt_dict,MainPg)],
		workers=[]
	   }.

%%----------------- JOIN_TAG -----------------------------------------
exp_fold({?JOIN_TAG,SubExprList,RecFun},Acc) when ?IS_RECFUN(RecFun) andalso is_record(Acc,acc) ->
	SenderSignList=[{_SS=make_ref(),SubExpr}||SubExpr<-SubExprList],
	JoinSpecs=[{SS,no_key_fun}||{SS,_}<-SenderSignList],
	FormatPipe=pipe_mapper:new(RecFun, Acc#acc.last_pipe),
	JPipe=pipe_join:new(JoinSpecs, FormatPipe),
%% 	wrap_log(pipe_join:new(JoinSpecs, FormatPipe),"just_before_join",
%% 				   fun()->
%% 						   io:format("DBG JOIN PARTIES ~p~n",[[SS||{SS,_}<-JoinSpecs]]),
%% 						   length(SubExprList)==4
%% 				   end),
	Acc1=add_group(_PgId=make_ref(), _Comm=join_bruteforce, Acc#acc{last_pipe=JPipe}),
	LastPipe=Acc1#acc.last_pipe,
	lists:foldl(
	  fun({SS,{?MAP_TAG,SubExpr,RecFun1}},AccIn) ->
			  exp_fold({?MAP_TAG,SubExpr,RecFun1,{'$SS',SS}},
					   AccIn#acc{last_pipe=LastPipe,pg=Acc1#acc.pg});
		 ({SS,SubExpr},AccIn) ->
			  exp_fold({?MAP_TAG,SubExpr,_RecFun=fun as_is/1,{'$SS',SS}},
					   AccIn#acc{last_pipe=LastPipe,pg=Acc1#acc.pg})
	  end, Acc1, SenderSignList);
%%----------------- JOIN_LKP_TAG -------------------------------------
exp_fold({?JOIN_LKP_TAG,CoreExpr,VectSpecList,RecFun},Acc) when ?IS_RECFUN(RecFun) ->
	LookupFunList=[
				   begin
					   case VectSpec of
						   {?VECT_TAG,V} ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN1 ~p~n", [Record]),
									   Key=element(2,Record),
									   iter_vect_query:new(V, {lookup,Key})
							   end;
						   {?VECT_TAG,V,VRecFun} when is_function(VRecFun,1) ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN2 ~p [~p,~p]~n", [Record,node(),self()]),
									   Key=element(2,Record),
									   BaseIter=iter_vect_query:new(V, {lookup,Key}),
									   iter_mapper:new(BaseIter, VRecFun)
							   end;
						   {?FILTER_TAG,{?VECT_TAG,V},FiltPredFun} ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN3 ~p~n", [Record]),
									   Key=element(2,Record),
									   iter_filter:new(iter_vect_query:new(V, {lookup,Key}), FiltPredFun)
							   end;
						   {?FILTER_TAG,{?VECT_TAG,V,VRecFun},FiltPredFun} when is_function(VRecFun,1) ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN4 ~p~n", [Record]),
									   Key=element(2,Record),
									   BaseIter=iter_vect_query:new(V, {lookup,Key}),
									   iter_filter:new(iter_mapper:new(BaseIter, VRecFun), FiltPredFun)
							   end;
						   BadVectSpec ->
							   ?LOG_ERR("VECT SPECS ~p~n", [VectSpecList]),
							   exit({unsupported_vect_spec,BadVectSpec})
					   end
				   end
				   ||VectSpec<-VectSpecList],
	FormatPipe=pipe_mapper:new(RecFun, Acc#acc.last_pipe),
	JoinLkpPipe=pipe_join_lookup:new(LookupFunList, ls, FormatPipe),
	exp_fold(CoreExpr,Acc#acc{last_pipe=JoinLkpPipe});

%%----------------- MAP_TAG ------------------------------------------
exp_fold({?MAP_TAG,SubExpr,RecFun,{'$SS',SS}},Acc) when ?IS_RECFUN(RecFun) ->
	Id=SS,
	MapredPipe=pipe_mapred1:new(Id,{pg_id,element(2,Acc#acc.pg)},Acc#acc.route_snap, Acc#acc.last_pipe),
	FormatPipe=pipe_mapper:new(RecFun, MapredPipe),
	Acc1=add_group(make_ref(),_Comm=ring_map_group,Acc),
	exp_fold(SubExpr,Acc1#acc{last_pipe=FormatPipe});
exp_fold({?MAP_TAG,SubExpr,RecFun},Acc) when ?IS_RECFUN(RecFun) ->
	exp_fold({?MAP_TAG,SubExpr,RecFun,{'$SS',nil}},Acc);
%%----------------- VECT_TAG -----------------------------------------
exp_fold({?VECT_TAG,Vect},Acc) ->
	exp_fold({?VECT_TAG,Vect,all},Acc);
exp_fold({?VECT_TAG,Vect,RecFun},Acc) when ?IS_RECFUN(RecFun) ->
	FormatPipe=pipe_mapper:new(RecFun, Acc#acc.last_pipe),
	exp_fold({?VECT_TAG,Vect,all},Acc#acc{last_pipe=FormatPipe});
exp_fold({?VECT_TAG,Vect,Query,RecFun},Acc)  when ?IS_RECFUN(RecFun) ->
	FormatPipe=pipe_mapper:new(RecFun, Acc#acc.last_pipe),
	exp_fold({?VECT_TAG,Vect,Query},Acc#acc{last_pipe=FormatPipe});
exp_fold({?VECT_TAG,Vect,{lookup,Key}=Query},Acc) when ?IS_VECT(Vect) ->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key, Acc#acc.route_snap)),
	Iter=iter_vect_query:new(Vect, Query),
 	pump_work(Node,Iter,_Comm=vect_lookup_pump,Acc);
exp_fold({?VECT_TAG,Vect,all},Acc) when ?IS_VECT(Vect) ->
	UpNodes=route_snap_to_nodes(Acc#acc.route_snap),
	Iter=iter_vect_query:new(Vect, all),
	lists:foldl(fun(Node,AccIn)->
						pump_work(Node,Iter,_Comm=vect_all_pump,AccIn)
				end, Acc, UpNodes);
%%----------------- RS_TAG -------------------------------------------
exp_fold({?RS_TAG,Iter}=_Expr,Acc) ->
	exp_fold({?RS_TAG,Iter,node()},Acc);
exp_fold({?RS_TAG,Iter,Node}=_Expr,Acc) ->
	pump_work(Node, Iter, _Comm=rs_pump, Acc);
%%----------------- FORMAT_TAG ---------------------------------------
exp_fold({?FORMAT_TAG,SubExpr,RecFun},Acc) when ?IS_RECFUN(RecFun) ->
	FormatPipe=pipe_mapper:new(RecFun, Acc#acc.last_pipe),
	exp_fold(SubExpr,Acc#acc{last_pipe=FormatPipe});
%%----------------- BAD_EXP ------------------------------------------
exp_fold(BadExpr,Acc) ->
	throw({bad_expr,BadExpr,Acc}).


%%====================================================================
%% Public utility functions
%%====================================================================
map_iter2pipe(Iter,Pipe,MapRecFun) ->
	MapExpr = {?MAP_TAG, 
				{?RS_TAG,Iter} , 
				MapRecFun},
	valg2pgt:fold(MapExpr, Pipe).
%%====================================================================
%% Internal functions
%%====================================================================

pump_work(Node,Iter,Comment,Acc) ->
 	Pump=pump:new(Iter, Acc#acc.last_pipe),
 	Worker=pump:to_pgt_worker(_Id={Comment, make_ref()},Pump,Node),
	add_worker(Worker,Acc).

add_worker(Worker,Acc) ->
	Acc#acc{pgt_dict=dict:append(Acc#acc.pg, Worker, Acc#acc.pgt_dict)}.

add_group(PgId,Comment,Acc) ->
	NewPg={pg,PgId,Comment},
	PgtDictNew=dict:append(Acc#acc.pg, NewPg , Acc#acc.pgt_dict),
	Acc#acc{pgt_dict=PgtDictNew,pg=NewPg}.

route_snap_to_nodes(NRouteSnap) ->
	gb_sets:to_list(
	  lists:foldl(fun(Route,AccIn)->
						  Node=?ROUTE2UPNODE(Route),
						  gb_sets:add(Node, AccIn)
				  end, 
				  gb_sets:new(), 
				  gb_trees:values(NRouteSnap))).

make_pgt(PgtDict,{pg,Id,Name}=RootPg) ->
%% 	?LOG_TRACE("pgt dict ~p~n",[dict:to_list(PgtDict)]),
%% 	exit(user),
	{PgProtos,Workers}=case dict:find(RootPg, PgtDict) of
		error ->
			{[],[]};
		{ok,ChildrenList} when is_list(ChildrenList) ->
			{[Pg||Pg<-ChildrenList,element(1,Pg)==pg],[Wrk||Wrk<-ChildrenList,is_record(Wrk, worker)]}
	end,
	#pg{id=Id,name=Name,pgs=[make_pgt(PgtDict,Pg)||Pg<-PgProtos],workers=Workers}.

format(Pg) ->
	pgt:explainex(Pg,fun(Tab,{Node,Module,Fun,_Args})->
							 lists:append([
										   io_lib:format(Tab++"Node=~p~n",[Node]),
							 			   io_lib:format(Tab++"Module=~p~n",[Module]),
							 			   io_lib:format(Tab++"Fun=~p~n",[Fun])])
				  end).
as_is(Rec) ->
	Rec.

wrap_log(Pipe,LogFileName,CondFun) ->
	case CondFun() of
		true ->
			pipe_logger:new(LogFileName, Pipe);
		_ ->
			Pipe
	end.
%%---------------------------------------------------------
%% Test Cases
%%---------------------------------------------------------	
test(map_rs) ->
	V1=vect_mnesia:new_mem_bag(ages1, [fname_lname,age]),
	Rs={?RS_TAG,iter:from_list([
									 {"Vasya","Ivanov",23},
									 {"Dima","Pupkin",27},
									 {"Masha","Bublikova",32},
									 {"Katya","Shkvarkina",19},
									 {"Yulya","Sharikova",34},
									 {"Vasya","Bukin",14},
									 {"Ostap","Bender",36},
									 {"Yuriy","Dolgorukiy",44},
									 {"Chingiz","Han",50}
									])},
	MapExpr1 = {?MAP_TAG, 
				Rs , 
				fun({Fn,Ln,Age}=_V)-> {ages1,{Fn,Ln},Age} end},
	Pg=fold(MapExpr1, pipe_vect:new(V1)),
 	io:format(format(Pg));

test(vect_join) ->
 	load_ages_hobbies(),
	V1=vect_mnesia:new_mem_bag(ages, [id,age]),
	V2=vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
	V3=vect_mnesia:new_mem_bag(j1, [fname,lname,age,hobby]),
	MapExpr1 = {?VECT_TAG,V1},
	MapExpr2 = {?VECT_TAG,V2},
	JoinExpr = {?JOIN_TAG,
				[MapExpr1,MapExpr2],
				fun([Key,{ages,{Fn,Ln},Age},{hobbies,{Fn,Ln},Hobby}])-> 
						{j1,Fn,Ln,Age,Hobby};
				   (X) ->
						exit({bag_out_format,X})
				end },
	Pg=fold(JoinExpr, pipe_console:new()),
%% 	io:format(format(Pg)),
	pgt:new(Pg);
test(map_join) ->
%%  	load_ages_hobbies(),
	V1=vect_mnesia:new_mem_bag(ages, [id,age]),
	V2=vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
	V3=vect_mnesia:new_mem_bag(j1, [fname,lname,age,hobby]),
	MapExpr1 = {?MAP_TAG,{?VECT_TAG,V1},
				fun(Rec)->
						Rec 
				end},
	MapExpr2 = {?MAP_TAG,{?VECT_TAG,V2},
				fun(Rec)->
						Rec 
				end},
	JoinExpr = {?JOIN_TAG,
				[MapExpr1,MapExpr2],
				fun([Key,{ages,{Fn,Ln},Age},{hobbies,{Fn,Ln},Hobby}])-> 
						{j1,Fn,Ln,Age,Hobby};
				   (X) ->
						exit({bag_out_format,X})
				end },
	Pg=fold(JoinExpr, pipe_console:new()),
%% 	io:format(format(Pg)),
	pgt:new(Pg);
test(map_map) ->
  	load_ages_hobbies(),
	V1=vect_mnesia:new_mem_bag(ages, [id,age]),
	V2=vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
	V3=vect_mnesia:new_mem_bag(j1, [fname,lname,age,hobby]),
	MapExpr1 = {?MAP_TAG,{?VECT_TAG,V1},
				fun({ages,Id,Age})->
						{ages,Id,Age}
				end},
	MapExpr2 = {?MAP_TAG,MapExpr1,
				fun({ages,Id,Age})->
						{sega,Age,Id} 
				end},
	MapExpr3 = {?MAP_TAG,MapExpr2,
				fun({sega,Age,Id}=Rec)->
						{ages_by_hash,erlang:phash(Rec,4),Id,Age} 
				end},
	MapExpr4 = {?MAP_TAG,MapExpr3,
				fun({ages_by_hash,Hash,Id,Age}=Rec)->
						{out,1,Rec} 
				end},
	Pg=fold(MapExpr4, pipe_console:new()),
 	io:format(format(Pg))
 ,	pgt:new(Pg)
.

load_ages_hobbies() ->
	map_from_file("../test/valg_SUITE_data/ages.tup",vect_mnesia:new_mem_bag(ages, [id,age]),
				  fun({X,Y,Z})->
						  {ages,{X,Y},Z}
				  end),
	map_from_file("../test/valg_SUITE_data/hobbies.tup",vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
				  fun({X,Y,Z})->
						  {hobbies,{X,Y},Z}
				  end),
	receive
		after 1000 ->
			ok
	end.


map_from_file(Filename,Vect,RecFun) ->
	{ok,Tuples}=file:consult(Filename),
	Rs={?RS_TAG,iter:from_list(Tuples)},
	MapExpr1 = {?MAP_TAG, 
				Rs , 
				RecFun},
	pgt:new(fold(MapExpr1, pipe_vect:new(Vect))).
